package com.heima.stream.listener;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Arrays;

//监听器
@EnableBinding( WordProcess.class)//绑定监听接口
public class WordListener {
    @StreamListener("word_1")//指定接收主题
    @SendTo("counts_1")//指定接受主题
    public KStream<String, String> process(KStream<String,String> input) {//参数   与   返回都是 Kstream
        //接收数据  把数据转换到集合中
        KStream<String, String> flatMapValues = input.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.toLowerCase().split("\\W+"));
            }
        });
        //把集合里面的数据分组
        KGroupedStream<String, String> groupedStream = flatMapValues.groupBy(new KeyValueMapper<String, String, String>() {
            @Override
            public String apply(String key, String value) {
                return value;
            }
        });
        //把分组的数据 再指定时间间隔接收
        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(6)));
        //把数据进行聚合处理  分组 累计
        KTable<Windowed<String>, Long> count = windowedKStream.count(Materialized.with(Serdes.String(), Serdes.Long()));//分组
        KStream<String, String> map = count.toStream().map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {//格式转换
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                return new KeyValue<>(key.key(), value.toString());
            }
        });
        return map;
    }
}
